package cn;

import java.nio.ByteBuffer;

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

public class Demo {

	public static void main(String[] args) {
		Demo m = new Demo();
		m.runJava();
	}

	int ITEMS = 25;
	int QUEUE_SIZE = 10000;
	
	private Context context;
	private Socket socket;

	private void runJava() {
		// Logger.debug("启动ZMQ接收进程...");
		this.context = ZMQ.context(1);
		this.socket = this.context.socket(ZMQ.PULL);
		this.socket.bind("tcp://127.0.0.1:7788");
		this.socket.setHWM(QUEUE_SIZE);
		// this.socket.setLinger(0);
		// Logger.debug("ZMQ接收进程启动成功，监听端口： " + Globals.RECV_DATA_PORT);

		// ByteBuffer buffer =
		// ByteBuffer.allocateDirect(1024).order(ByteOrder.nativeOrder());
		ByteBuffer buffer = ByteBuffer.allocate(10240);
		System.out.println("start....");
		byte[] bs = buffer.array();
		int l = 0;
		int count = 0;
		while (true) {
			buffer.position(0);

			// String s = this.socket.recvStr();
			String str = "";
			for (int k = 0; k < ITEMS; k++) {
				buffer.position(0);

				this.socket.recvByteBuffer(buffer, 0);
				int last = buffer.position();
				if (-1 == this.socket.recvByteBuffer(buffer, 0)) {
					System.out.println("error " + buffer.position());
				} else {
					str = new String(bs, last, buffer.position());
					// System.out.println(str);

					this.socket.recvByteBuffer(buffer, 0);
					last = buffer.position();
					l = this.socket.recvByteBuffer(buffer, 0);
					str = new String(bs, last, buffer.position());
					// System.out.println(str);

				}

				///
			}

			// System.out.println(str);
			this.socket.recvByteBuffer(buffer, 0);
			int readc = buffer.position();
			// System.out.println(readc);

			// System.out.println(s);
			// System.out.println(l);
			System.out.println("get:" + count++);
			// System.out.println("");
		}
	}
}
